/**
 * Free software.
 */

package org.nanhill.commons.pattern.dispatcher;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * Classic producter - consumer queue.
 * 
 * @author alin [xalinx at gmail dot com]
 * @date 2006-11-4
 */
public abstract class ListDispatcher<E> implements Dispatcher<E> {
	private final transient Log log = LogFactory.getLog(this.getClass());

	/**
	 * Default interval time between detects
	 */
	private static final long DEFAULT_QUEUE_DETECT_TIME = 10 * 1000L;

	/**
	 * Interval time between detects
	 */
	private long queueDetectTime = DEFAULT_QUEUE_DETECT_TIME;

	/**
	 * Work queue.
	 */
	private final List<E> queue = new LinkedList<E>();

	/**
	 * Queue lock.
	 */
	private final Lock queueLock = new ReentrantLock();

	/**
	 * Queue lock condition.
	 */
	private final Condition queueCond = queueLock.newCondition();

	/**
	 * Queue working stop flag.
	 */
	private volatile boolean stopped = false;

	protected ListDispatcher() {
	}

	protected ListDispatcher(long queueDetectTime) {
		this.queueDetectTime = queueDetectTime;
	}

	public final void incept(E product) {
		try {
			queueLock.lock();
			queue.add(product);
			queueCond.signal();
		} finally {
			queueLock.unlock();
		}

	}

	public void start() {
		new WorkerThread().start();
	}

	public final void stop() {
		try {
			queueLock.lock();
			stopped = true;
			queueCond.signal();
		} finally {
			queueLock.unlock();
		}
	}

	private class WorkerThread extends Thread {
		public void run() {
			while (true) {
				E product = null;
				try {
					queueLock.lock();
					while (queue.isEmpty() && !stopped) {
						boolean sleepWell = false;
						try {
							sleepWell = queueCond.await(queueDetectTime,
									TimeUnit.MILLISECONDS);
							if (log.isDebugEnabled()) {
								if (sleepWell) {
									log.debug("Sleep well!");
								} else {
									log.debug("Sleep poorly!");
								}
							}
						} catch (InterruptedException e) {
							if (log.isErrorEnabled()) {
								log.error("Interrupted!", e);
							}
							continue;
						}
					}
					if (stopped)
						return;
					product = queue.remove(0);
				} finally {
					queueLock.unlock();
				}
				send(product);
			}
		}
	}
}
